package com.itao.rocketmq5.util;

import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.producer.Producer;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @Author sjt
 * @CreateTime 2023/10/10 9:51
 * message.type取值为 NORMAL| FIFO | DELAY | TRANSACTION 默认为NORMAL
 * sh mqadmin updateTopic -n 192.168.56.128:9876  -t normal_topic -c DefaultCluster -a +message.type=NORMAL
 */
public class RocketMqFactory {

    private final static Map<String, Producer> PRODUCERS = new ConcurrentHashMap<>();
    // 接入点地址，需要设置成Proxy的地址和端口列表，一般是xxx:8081;xxx:8081。
    private static final String endpoint = "192.168.56.128:8081";
    public static Producer getProducer(String topic) {
        var producer = PRODUCERS.get(topic);
        if (producer == null) {
            synchronized (RocketMqFactory.class) {
                producer = PRODUCERS.get(topic);
                if (producer == null) {
                    try {
                        ClientServiceProvider provider = provider();
                        producer = provider.newProducerBuilder()
                                .setTopics(topic)
                                .setClientConfiguration(configuration())
                                .build();
                        PRODUCERS.put(topic, producer);
                        return producer;
                    } catch (ClientException e) {
                        throw new RuntimeException(e);
                    }
                }
            }
        }
        return producer;
    }

    public static ClientServiceProvider provider(){
        return ClientServiceProvider.loadService();
    }

    public static ClientConfiguration configuration(){
        return ClientConfiguration
                .newBuilder()
                .setEndpoints(endpoint)
                .build();
    }
}
